package com.pactera.rabbitmq.service.consumer;

import com.alibaba.fastjson.JSON;
import com.pactera.rabbitmq.bean.MessageInfo;
import com.pactera.rabbitmq.utils.AdminUtils;
import com.rabbitmq.client.Channel;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 默认实时数据订阅类
 * 消费消息失败后，将消息推送到重试队列，重试3次后将消息推送至失败队列
 *
 * @author: liqingdian
 **/
public class RtdsDefaultSubscriber extends RtdsSubscriber {

    private static final Logger log = LoggerFactory.getLogger(RtdsDefaultSubscriber.class);

    private RtdsAccept rtdsAccept;
    private RabbitTemplate rabbitTemplate;

    public RtdsDefaultSubscriber(RtdsAccept rtdsAccept, RabbitTemplate rabbitTemplate) {
        this.rtdsAccept = rtdsAccept;
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        this.recevice(message);
    }

    @Override
    public void recevice(Message message) {
        byte[] body = message.getBody();
        if (ArrayUtils.isNotEmpty(body)) {
            try {
                MessageInfo messageInfo = JSON.parseObject(new String(body, StandardCharsets.UTF_8), MessageInfo.class);
                rtdsAccept.accept(messageInfo);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
                MessageProperties properties = message.getMessageProperties();
                String queue = properties.getConsumerQueue();
                long retryCount = getRetryCount(properties);
                if (retryCount <= 3) {
                    log.info("处理失败，将消息发送至重试队列，当前重试次数: {}", retryCount);
                    Map<String, Object> headers = properties.getHeaders();
                    if (headers == null) {
                        headers = new HashMap<>();
                    }
                    headers.put("x-orig-routing-key", getOrigRoutingKey(properties, queue));
                    Message sendMsg = new Message(message.getBody(), createOverrideProperties(properties, headers));
                    rabbitTemplate.convertAndSend(AdminUtils.retry(AdminUtils.EXCHANGE_RTDS), AdminUtils.retry(queue), sendMsg);
                } else {
                    log.info("处理失败，将消息发送至失败队列");
                    Map<String, Object> headers = properties.getHeaders();
                    headers.put("x-orig-routing-key", getOrigRoutingKey(properties, queue));
                    Message sendMsg = new Message(message.getBody(), createOverrideProperties(properties, headers));
                    rabbitTemplate.convertAndSend(AdminUtils.failed(AdminUtils.EXCHANGE_RTDS), AdminUtils.failed(AdminUtils.QUEUE_RTDS), sendMsg);
                }
            }
        }
    }

    /**
     * 获取消息重试次数
     *
     * @param properties AMQP消息属性
     * @return 消息重试次数
     */
    public Long getRetryCount(MessageProperties properties) {
        Long retryCount = 0L;
        try {
            Map<String, Object> headers = properties.getHeaders();
            if (headers != null) {
                if (headers.containsKey("x-death")) {
                    List<Map<String, Object>> deaths = (List<Map<String, Object>>) headers.get("x-death");
                    if (deaths.size() > 0) {
                        Map<String, Object> death = deaths.get(0);
                        retryCount = (Long) death.get("count");
                    }
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return retryCount;
    }

    /**
     * 获取原始的routingKey
     *
     * @param properties   AMQP消息属性
     * @param defaultValue 默认值
     * @return 原始的routing-key
     */
    public String getOrigRoutingKey(MessageProperties properties, String defaultValue) {
        String routingKey = defaultValue;
        try {
            Map<String, Object> headers = properties.getHeaders();
            if (headers != null) {
                if (headers.containsKey("x-orig-routing-key")) {
                    routingKey = headers.get("x-orig-routing-key").toString();
                }
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
        return routingKey;
    }

    /**
     * 从已有的properties中创建新的properties，使用提供的headers字段覆盖已有的headers
     *
     * @param properties AMQP属性
     * @param headers    要覆盖的headers
     * @return 新创建的properties
     */
    public MessageProperties createOverrideProperties(MessageProperties properties, Map<String, Object> headers) {
        MessageProperties newPro = new MessageProperties();
        newPro.setConsumerQueue(properties.getConsumerQueue());
        newPro.setMessageId(properties.getMessageId());
        newPro.setType(properties.getType());
        newPro.setTimestamp(properties.getTimestamp());
        newPro.setUserId(properties.getUserId());
        newPro.setAppId(properties.getAppId());
        newPro.setClusterId(properties.getClusterId());
        newPro.setExpiration(properties.getExpiration());
        newPro.setReplyTo(properties.getReplyTo());
        newPro.setContentType(properties.getContentType());
        newPro.setContentEncoding(properties.getContentEncoding());
        newPro.setDeliveryMode(properties.getDeliveryMode());
        newPro.setPriority(properties.getPriority());
        newPro.setCorrelationId(properties.getCorrelationId());
        if (MapUtils.isNotEmpty(headers)) {
            for (String key : headers.keySet()) {
                newPro.setHeader(key, headers.get(key));
            }
        }
        return newPro;
    }
}
